package com.sfx.shortLink.mq.consumer;


import com.sfx.shortLink.common.constants.RedisKeyConstant;
import com.sfx.shortLink.common.convention.exception.ServiceException;
import com.sfx.shortLink.dto.ShortLinkStatsRecordDTO;
import com.sfx.shortLink.mq.idempotent.MessageIdempotentHandler;
import com.sfx.shortLink.service.impl.ShortLinkServiceImpl;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.LockSupport;

@RequiredArgsConstructor
@Slf4j
@Component
public class ShortLinkStatsDelayQueueConsumer implements InitializingBean {

    private static final AtomicInteger index = new AtomicInteger();

    private final RedissonClient redissonClient;
    private final MessageIdempotentHandler messageIdempotentHandler;
    private final ShortLinkServiceImpl shortLinkService;

    public void onMessage() {
        Executors.newSingleThreadExecutor(r -> {
            Thread t = new Thread(r);
            t.setName("ShortLinkStatsDelayQueueConsumer-" + index.getAndDecrement());
            t.setDaemon(Boolean.TRUE);
            return t;
        }).execute(() -> {
            for (; ; ) {
                try {
                    RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(RedisKeyConstant.DELAY_QUEUE_STATS_KEY);
                    RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
                    ShortLinkStatsRecordDTO statsRecordDTO = (ShortLinkStatsRecordDTO) delayedQueue.poll();
                    String keys = Objects.requireNonNull(statsRecordDTO).getKeys();
                    if (!messageIdempotentHandler.isConsume(keys)) {
                        if (messageIdempotentHandler.isAccomplish(keys)) {
                            break;
                        }
                        throw new ServiceException("消息消费处理出现异常,无法被消费");
                    }
                    try {
                        shortLinkService.shortLinkStats(null, null, statsRecordDTO);
                    } catch (Throwable throwable) {
                        messageIdempotentHandler.delMessage(keys);
                        log.error("延迟消费处理异常", throwable);
                    }
                    // 消息消费已完成
                    messageIdempotentHandler.setAccomplish(keys);
                    continue;
                } catch (Throwable throwable) {
                }
                LockSupport.parkUntil(500);
            }
        });
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        onMessage();
    }
}
